package com.gome.ss.core.schedule;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.bft.BitFetchTable;
import com.gome.spiderkit.client.WebPage;
import com.gome.ss.common.AppConfiguer;
import com.gome.ss.core.download.SpiderRequest;
import com.gome.ss.core.download.SpiderRequestInterface;
import com.gome.ss.core.plugin.PluginManager;
import com.gome.ss.core.queue.QueueCacheInterface;
import com.gome.ss.core.queue.RequestQueue;

public class ScheduleCore {
	
	/*
	 * 
	private ConcurrentHashMap<String, Integer> taskTotailReferenceMap = new ConcurrentHashMap<String, Integer>();
	private ConcurrentHashMap<String, Integer> taskReferenceInThreadPoolMap = new ConcurrentHashMap<String, Integer>();
	private ConcurrentHashMap<String, Integer> pluginProcessReferenceMap = new ConcurrentHashMap<String, Integer>();
	 * 
	 */
	
	private static Logger logger = Logger.getLogger(ScheduleCore.class);
	
	private Map<String, Integer> taskTotailReferenceMap = new HashMap<String, Integer>();
	private Map<String, Integer> taskReferenceInThreadPoolMap = new HashMap<String, Integer>();
	private Map<String, Integer> pluginProcessReferenceMap = new HashMap<String, Integer>();
	
	private List<String> needStopTaskIDs = new ArrayList<String>();
	private Queue<SpiderRequestInterface> scheduleQueue = new LinkedList<SpiderRequestInterface>();
	private BitFetchTable fetchTable = null;
	private RequestQueue requestQueue = null;
	private ThreadTaskPool pool = null;
	private ThreadSpecialTaskPool spool = null;
	private ScheduleTask scheduleTask = null;
	private int scheduleInterval = -1;
	private Timer timer = new Timer();
	
	public void init(String pluginName, int threadCount, int scheduleInterval) {
		
		this.scheduleInterval = scheduleInterval;
		
		if (fetchTable == null) {
			fetchTable = new BitFetchTable("spider-schedule-"+pluginName);
			fetchTable.createMemTable();
			requestQueue = new RequestQueue();
		}
		
		if (pool == null) {
			pool = new ThreadTaskPool(this, threadCount);
			pool.init();
		}
		
		if (spool == null) {
			spool = new ThreadSpecialTaskPool(this, threadCount);
			spool.init();
		}
		
		if (scheduleTask == null) {
			scheduleTask = new ScheduleTask(this);
			if (this.scheduleInterval < 0)
				this.scheduleInterval = AppConfiguer.getIntByKey("scheduleInterval");
			
			timer.schedule(scheduleTask, 500, this.scheduleInterval);
		}
	}
	
	public void destory() {
		timer.cancel();
		fetchTable.destoryTable();
	}
	
	public int getScheduleInterval() {
		return scheduleInterval;
	}

	public int pluginReference(String taskID) {
		synchronized(ScheduleCore.class) {
			if (pluginProcessReferenceMap.get(taskID) == null)
				return 0;
			return pluginProcessReferenceMap.get(taskID);
		}
	}
	
	private void appendReferenceInQueue(String taskID) {
		synchronized(ScheduleCore.class) {
			if (pluginProcessReferenceMap.containsKey(taskID)) {
				pluginProcessReferenceMap.put(taskID, pluginProcessReferenceMap.get(taskID) + 1);
			} else {
				pluginProcessReferenceMap.put(taskID, 1);
			}	
		}
	}
	
	public void addTaskReferenceInThreadPool(String taskID) {
		synchronized(ScheduleCore.class) {
			if (taskTotailReferenceMap.containsKey(taskID)) {
				taskTotailReferenceMap.put(taskID, taskTotailReferenceMap.get(taskID) + 1);
			} else {
				taskTotailReferenceMap.put(taskID, 1);
			}
			
			if (taskReferenceInThreadPoolMap.containsKey(taskID)) {
				taskReferenceInThreadPoolMap.put(taskID, taskReferenceInThreadPoolMap.get(taskID) + 1);
			} else {
				taskReferenceInThreadPoolMap.put(taskID, 1);
			}
			
			if (pluginProcessReferenceMap.containsKey(taskID)) {
				pluginProcessReferenceMap.put(taskID, pluginProcessReferenceMap.get(taskID) - 1);
				if (pluginProcessReferenceMap.get(taskID) <= 0) {
					pluginProcessReferenceMap.remove(taskID);
				}
			}
		}
	}
	
	public void removeTaskInThreadPool(String taskID) {
		synchronized(ScheduleCore.class) {
			if (taskReferenceInThreadPoolMap.containsKey(taskID)) {
				taskReferenceInThreadPoolMap.put(taskID, taskReferenceInThreadPoolMap.get(taskID) - 1);
				if (taskReferenceInThreadPoolMap.get(taskID) <= 0) {
					taskReferenceInThreadPoolMap.remove(taskID);
				}
			}
		}
	}
	
	public int taskReferenceInThreadPool(String taskID) {
		synchronized(ScheduleCore.class) {
			if (taskReferenceInThreadPoolMap.get(taskID) == null)
				return 0;
			return taskReferenceInThreadPoolMap.get(taskID);
		}
	}
	
	public int taskTotalReference(String taskID) {
		synchronized(ScheduleCore.class) {
			if (taskTotailReferenceMap.get(taskID) == null)
				return 0;
			return taskTotailReferenceMap.get(taskID);
		}
	}
	
	public void cleanTaskTotalReference(String taskID) {
		synchronized(ScheduleCore.class) {
			if (taskTotailReferenceMap.containsKey(taskID))
				taskTotailReferenceMap.remove(taskID);
		}
	}
	
	public int taskReferenceInRequestQueuel(String taskID) {
		synchronized(ScheduleCore.class) {
			return pluginReference(taskID);
		}
	}
	
	public int taskReference(String taskID) {
		synchronized(ScheduleCore.class) {
			int ref = taskReferenceInThreadPool(taskID) + taskReferenceInRequestQueuel(taskID);
			return ref;
		}
	}
	
	public void schedule() {
		synchronized(ScheduleCore.class) {
			SpiderRequestInterface req = requestQueue.poll();
			
			while (req != null && taskNeedStop(req.getSourceTaskID()))
				req = requestQueue.poll();
			
			if (req != null) {
				scheduleQueue.add(req);
			}
		}
	}
	
	public void addToStopTaskList(String taskID) {
		needStopTaskIDs.add(taskID);
	}
	
	public boolean taskNeedStop(String taskID) {
		return needStopTaskIDs.contains(taskID);
	}
	
	public void pushRequestToQueue(SpiderRequestInterface obj, Integer priority) {
		pushRequestToQueue(obj, priority, true);
	}
	
	public synchronized void pushRequestToQueue(SpiderRequestInterface obj, Integer priority, boolean filte) {
		
		if (obj == null)
			return;
		
		if (obj.getIdentity() == null)
			return;
		
		if (taskNeedStop(obj.getSourceTaskID())) {
			if (obj.isPageReady()) {
				try {
					SpiderRequest r = (SpiderRequest) obj;
					if (r.getReadyPage() != null)
						r.getReadyPage().destory(true);
				} catch (Exception e) { }
			}
			return;
		}
		
		if (filte) {
			synchronized (RequestQueue.class) {
				if (fetchTable.testBit(obj.getIdentity()))
					return;
				fetchTable.setBit(obj.getIdentity());
			}
		}
		
		appendReferenceInQueue(obj.getSourceTaskID());
		
		if (obj.isPageReady())
			requestQueue.add(obj, QueueCacheInterface.specialPriority);
		else
			requestQueue.add(obj, priority);
	}
	
	public SpiderRequestInterface pollRequestFromQueue() {
		synchronized(ScheduleCore.class) {
			SpiderRequestInterface req = scheduleQueue.poll();
			while (req != null && taskNeedStop(req.getSourceTaskID()))
				req = requestQueue.poll();
			return req;
		}
	}
	
	public synchronized SpiderRequestInterface pollRequestFromSpecialQueue() {
		return requestQueue.pollFromSpecial();
	}
	
	private class ThreadTaskPool {
		private int threadCount = 0;
		private ExecutorService executorService = null;
		private ScheduleCore scheduleCore = null;
		
		public ThreadTaskPool(ScheduleCore sc, int tc) {
			scheduleCore = sc;
			threadCount = tc;
			executorService = Executors.newFixedThreadPool(tc);
		}
		
		public void init() {
			for (int i = 0; i < threadCount; i ++) {
				RequestThreadTask task = new RequestThreadTask(scheduleCore);
				Thread thread = ThreadTaskBuilder.newThreadTask(task);
				executorService.execute(thread);
			}
		}
		
		public void shutdown() {
			executorService.shutdown();
		}
	}
	
	private class ThreadSpecialTaskPool {
		private int threadCount = 0;
		private ExecutorService executorService = null;
		private ScheduleCore scheduleCore = null;
		
		public ThreadSpecialTaskPool(ScheduleCore sc, int tc) {
			threadCount = tc;
			scheduleCore = sc;
			executorService = Executors.newFixedThreadPool(tc);
		}
		
		public void init() {
			for (int i = 0; i < threadCount; i ++) {
				RequestThreadTask task = new RequestThreadTask(scheduleCore, true);
				Thread thread = ThreadTaskBuilder.newThreadTask(task);
				executorService.execute(thread);
			}
		}
		
		public void shutdown() {
			executorService.shutdown();
		}
	}
	
	private class ScheduleTask extends TimerTask {
		private ScheduleCore sc = null;
		
		public ScheduleTask(ScheduleCore sc) {
			this.sc = sc;
		}
		
		@Override
		public void run() {
			// TODO Auto-generated method stub
			sc.schedule();
		}
	}
}
